package ntu.p2p.dht;

import ntu.p2p.dht.messaging.DHTDifferenceDeserializer;
import ntu.p2p.dht.messaging.DHTFetchMessage;
import ntu.p2p.dht.messaging.DHTLookupMessage;
import ntu.p2p.dht.messaging.DHTObjectDifference;
import ntu.p2p.dht.messaging.InsertObjectMessage;
import ntu.p2p.dht.messaging.DHTUpdateObjectMessage;
import ntu.p2p.past.PastImpl;
import ntu.p2p.past.UpdateApplicator;
import ntu.p2p.past.messaging.ObjectDifference;
import ntu.p2p.past.messaging.UpdateObjectMessage;
import rice.Continuation;
import rice.Continuation.NamedContinuation;
import rice.Continuation.StandardContinuation;
import rice.environment.Environment;
import rice.environment.logging.Logger;
import rice.p2p.commonapi.Endpoint;
import rice.p2p.commonapi.Id;
import rice.p2p.commonapi.Message;
import rice.p2p.commonapi.Node;
import rice.p2p.commonapi.NodeHandle;
import rice.p2p.past.PastContent;
import rice.p2p.past.PastContentHandle;
import rice.p2p.past.PastPolicy;
import rice.p2p.past.PastPolicy.DefaultPastPolicy;
import rice.p2p.past.messaging.CacheMessage;
import rice.p2p.past.messaging.PastMessage;
import rice.persistence.StorageManager;

public class DHTImpl implements DHT {

	private class PrivatePastImpl extends PastImpl {

		public PrivatePastImpl(Node node, StorageManager manager, int replicas,
				String instance) {
			super(node, manager, replicas, instance, null);
		}

		public PrivatePastImpl(Node node, StorageManager manager, int replicas,
				String instance, PastPolicy policy) {
			super(node, manager, replicas, instance, policy, null);
		}
		
		@Override
		public void deliver(Id id, Message message) {
			//System.out.println(message.getClass().getCanonicalName());
			if (message instanceof UpdateObjectMessage) {
				final UpdateObjectMessage updateObject = (UpdateObjectMessage) message;
				ObjectDifference diff = updateObject.getDifference();
				/// deserialize diff
				if (diff instanceof InsertObjectMessage) {
					this.handleInsertObjectMessage((InsertObjectMessage) diff);
				} else if (diff instanceof DHTUpdateObjectMessage) {
					this.handleUpdateObjectMessage((DHTUpdateObjectMessage) diff);
				} else {
					super.deliver(id, message);
				}
			} else if (((PastMessage) message).isResponse()) {
				super.deliver(id, message);
	    } else if (message instanceof DHTLookupMessage) {
				this.handleDHTLookupMessage((DHTLookupMessage) message);
			} else if (message instanceof DHTFetchMessage) {
				this.handleDHTFetchMessage ((DHTFetchMessage) message);
			} else {
				super.deliver(id, message);
			}
		}

		@SuppressWarnings("unchecked")
		private void handleDHTFetchMessage(final DHTFetchMessage fmsg) {
      lookups++;

      Continuation c;
//      c = getResponseContinuation(msg);
      c = getFetchResponseContinuation(fmsg); // has to be special to determine how to send the message
      
      storage.getObject(fmsg.getHandle().getId(), new StandardContinuation(c) {

				@Override
				public void receiveResult(Object result) {
					if (result != null) {
						parent.receiveResult(((DHTContents) result).getDHTContents().get(fmsg.getKey()));
					} else {
						parent.receiveResult(null);
					}
				}
      	
      });
		}

		private void handleInsertObjectMessage(final InsertObjectMessage message) {
			System.out.println("insert dht object...........");
			final DHTContent content = message.getContent();
			this.lookupHandle(content.getKey().getId(), this.getLocalNodeHandle(),
					new Continuation<PastContentHandle, Exception>() {

				@Override
				public void receiveException(Exception exception) {
					// TODO Auto-generated method stub

				}

				@SuppressWarnings("unchecked")
				@Override
				public void receiveResult(PastContentHandle result) {
					inserts++;
					final Id objectId = message.getContent().getKey().getId();
					lockManager.lock(objectId, new StandardContinuation(this) {
						@Override
						public void receiveResult(Object result) {
							storage.getObject(objectId, new StandardContinuation(parent) {

								@Override
								public void receiveResult(Object result) {
									DHTContents dhtcontents;
									if (result == null) {
										dhtcontents = new DHTContents(content.getKey().getId());
									} else {
										dhtcontents = (DHTContents) result;
										if (dhtcontents.contains(content)) {
											parent.receiveException(new Exception() {

												private static final long serialVersionUID = -5615400848558267817L;
												
											});
										}
									}
									dhtcontents.getDHTContents().put(content.getKey(), content);
									storage.store(objectId, null, dhtcontents,
											new StandardContinuation(parent) {
										public void receiveResult(Object result) {
											lockManager.unlock(objectId);
										}
									});
								}
							});
						}
					});
				}
			});
		}
		
		private void handleUpdateObjectMessage(final DHTUpdateObjectMessage diff) {
			System.out.println("update dht object ...........");
			final DHTObjectDifference difference = diff.getDifference();
			this.lookupHandle(diff.getKey().getId(), this.getLocalNodeHandle(),
					new Continuation<PastContentHandle, Exception>() {

				@Override
				public void receiveException(Exception exception) {
					// TODO Auto-generated method stub

				}

				@SuppressWarnings("unchecked")
				@Override
				public void receiveResult(PastContentHandle result) {
					if (result == null) {
						this.receiveException(new Exception() { });
					}
					inserts++;
					final Id objectId = result.getId();
					lockManager.lock(objectId, new StandardContinuation(this) {
						@Override
						public void receiveResult(Object result) {
							storage.getObject(objectId, new StandardContinuation(parent) {

								@Override
								public void receiveResult(Object result) {
									DHTContents dhtcontents = (DHTContents) result;
									if (!dhtcontents.getDHTContents().containsKey(diff.getKey())) {
										parent.receiveException(new Exception() { });
										return;
									}
									DHTContent content = 
											dhtcontents.getDHTContents().get(diff.getKey());
									content = dhtUpdateApplicator.applyUpdate(content, difference);
									dhtcontents.getDHTContents().put(diff.getKey(), content);
									storage.store(objectId, null, dhtcontents,
											new StandardContinuation(parent) {
										public void receiveResult(Object result) {
											lockManager.unlock(objectId);
										}
									});
								}
							});
						}
					});
				}
			});
		}
		
		@SuppressWarnings("unchecked")
		private void handleDHTLookupMessage(final DHTLookupMessage lmsg) {
      lookups++;
      
      // if the data is here, we send the reply, as well as push a cached copy
      // back to the previous node
      storage.getObject(lmsg.getId(), new StandardContinuation(getResponseContinuation(lmsg)) {
        public void receiveResult(Object o) {
          if (logger.level <= Logger.FINE) logger.log("Received object " + o + " for id " + lmsg.getId());

          DHTContent result = null;
          if (o != null) {
          	result = ((DHTContents) o).getDHTContents().get(lmsg.getKey());
          }
          // send result back
          parent.receiveResult(result);
          
          /*
          // if possible, pushed copy into previous hop cache
          if ((lmsg.getPreviousNodeHandle() != null) &&
              (o != null) &&
              (! ((PastContent) o).isMutable())) {
            NodeHandle handle = lmsg.getPreviousNodeHandle();
            if (logger.level <= Logger.FINE) logger.log("Pushing cached copy of " + ((PastContent) o).getId() + " to " + handle);
            
            CacheMessage cmsg = new CacheMessage(getUID(), (PastContent) o, getLocalNodeHandle(), handle.getId());    
            //endpoint.route(null, cmsg, handle);
          }
          */
        }
      });
		}
		
	  @SuppressWarnings("unchecked")
		public void lookup(final DHTContentKey key, final boolean cache, final Continuation command) {
	    final Id id = key.getId();
	    
	    storage.getObject(id, new StandardContinuation(command) {
	      public void receiveResult(Object o) {
	        if (o != null) {
	          command.receiveResult(((DHTContents) o).getDHTContents().get(key));
	        } else {
	          // send the request across the wire, and see if the result is null or not
	          sendRequest(id, new DHTLookupMessage(getUID(), key, getLocalNodeHandle(), id), 
	          		new NamedContinuation("LookupMessage for " + id, this) {
	            public void receiveResult(final Object o) {
	              // if we have an object, we return it
	              // otherwise, we must check all replicas in order to make sure that
	              // the object doesn't exist anywhere
	              if (o != null) {
	    	          command.receiveResult(((DHTContents) o).getDHTContents().get(key));
	              } else {
	                lookupHandles(id, replicationFactor+1, new Continuation() {
	                  public void receiveResult(Object o) {
	                    PastContentHandle[] handles = (PastContentHandle[]) o;

	                    for (int i=0; i<handles.length; i++) {
	                      if (handles[i] != null) {
	                        fetch(handles[i], new StandardContinuation(parent) {
	                          public void receiveResult(final Object o) {
	                            // TODO fix this
	                          	command.receiveResult(((DHTContents) o).getDHTContents().get(key));
	                          }
	                        });
	                        
	                        return;
	                      }
	                    }

	                    // there were no replicas of the object
	                    command.receiveResult(null);
	                  }
	                  
	                  public void receiveException(Exception e) {
	                    command.receiveException(e);
	                  }
	                });
	              }
	            }
	            
	            public void receiveException(Exception e) {
	              // If the lookup message failed , we then try to fetch all of the handles, just
	              // in case.  This may fail too, but at least we tried.
	              receiveResult(null);
	            }
	          });
	        }
	      }
	    });
	  }
	  

		public void fetch(DHTContentHandle handle, Continuation command) {
	    PastContentHandle han = handle.getContentHandle();
	    sendRequest(han.getId(), new DHTFetchMessage(getUID(), handle, getLocalNodeHandle(), handle.getKey().getId()), 
	                new NamedContinuation("FetchMessage to " + handle.getContentHandle() + " for " + handle.getKey().getId(), command));
		}
	}

	private PrivatePastImpl past;
	private DHTUpdateApplicator dhtUpdateApplicator;
	private DHTContentDeserializer contentDeserializer;
	
	public DHTImpl(Node node, StorageManager manager, int replicas,
			String instance, DHTUpdateApplicator updateApplicator) {
		this.past = new PrivatePastImpl(node, manager, replicas, instance,
				new DefaultPastPolicy());
		this.dhtUpdateApplicator = updateApplicator;
	}

	public DHTImpl(Node node, StorageManager manager, int replicas,
			String instance, PastPolicy policy, DHTUpdateApplicator updateApplicator) {
		this.past = new PrivatePastImpl(node, manager,replicas, instance,	policy);
		this.dhtUpdateApplicator = updateApplicator;
	}

	@Override
	public void fetch(DHTContentHandle handle, Continuation command) {
		this.past.fetch(handle, command);
	}

	@Override
	public Environment getEnvironment() {
		return this.past.getEnvironment();
	}

	@Override
	public String getInstance() {
		return this.past.getInstance();
	}

	@Override
	public NodeHandle getLocalNodeHandle() {
		return this.past.getLocalNodeHandle();
	}

	@Override
	public int getReplicationFactor() {
		return this.past.getReplicationFactor();
	}

	@Override
	public void insert(DHTContent obj, Continuation<Boolean[], Exception> command) {
		UpdateObjectMessage updateMessage = new UpdateObjectMessage(
				obj.getKey().getId(), new InsertObjectMessage(obj));
		this.past.update(obj.getKey().getId(), updateMessage, command);
	}

	@Override
	public void lookup(DHTContentKey key,
			Continuation<DHTContent, Exception> command) {
		this.past.lookup(key, false, command);
	}

	@Override
	public void lookup(DHTContentKey key, boolean cache, Continuation command) {
		this.past.lookup(key, cache, command);
	}

	@Override
	public void lookupHandle(DHTContentKey key, NodeHandle handle,
			Continuation command) {
		this.past.lookupHandle(key.getId(), handle, command);
	}

	@Override
	public void lookupHandles(DHTContentKey key, int max, Continuation command) {
		this.past.lookupHandles(key.getId(), max, command);
	}

	@Override
	public void setContentDeserializer(DHTContentDeserializer deserializer,
			DHTDifferenceDeserializer differenceDeserializer) {
		this.contentDeserializer = deserializer;
		PastObjectDifferenceDeserializer temp = new PastObjectDifferenceDeserializer(deserializer, 
				differenceDeserializer);
		UpdateObjectMessage.setDifferenceDeserializer(temp);
	}

	@Override
	public void setContentHandleDeserializer(
			DHTContentHandleDeserializer deserializer) {
		// TODO Auto-generated method stub

	}

	@Override
	public void update(DHTContentKey key, DHTUpdateObjectMessage update,
			Continuation command) {
		UpdateObjectMessage updateMessage = 
				new UpdateObjectMessage(key.getId(), update);
		this.past.update(key.getId(), updateMessage, command);
	}
	
	@Override
	public Endpoint getEndpoint() {
		return this.past.getEndpoint();
	}
}
